package com.yet.container.remote;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.yet.dto.Request;
import com.yet.dto.Response;
import com.yet.exception.RemotingException;

/**
 * 回调结果处理
 */
public class ResponseFuture implements Future<Object> {

	private static final Logger logger = LoggerFactory.getLogger(ResponseFuture.class);
	private static final Map<String, ResponseFuture> FUTURES = new ConcurrentHashMap<String, ResponseFuture>();
	public static final int DEFAULT_TIMEOUT = 2000;
	private final Request request;
	private volatile Response response;
	private final String id;
	private AtomicInteger state = new AtomicInteger(0);// ready ok cancel
	private ReentrantLock lock = new ReentrantLock();
	private final Condition done = lock.newCondition();
	private volatile ResponseCallback callback;

	public ResponseFuture(Request request) {
		this.request = request;
		this.id = request.getId();
		FUTURES.put(id, this);
	}

	@Override
	public boolean cancel(boolean mayInterruptIfRunning) {
		Response errorResult = new Response(id);
		errorResult.setThrowable(new RemotingException("request future has been canceled."));
		response = errorResult;
		FUTURES.remove(id);
		state.getAndSet(3);
		return true;
	}

	@Override
	public boolean isCancelled() {
		return state.get() == 2;
	}

	@Override
	public boolean isDone() {
		return response != null && state.get() == 1;
	}

	public static void received(Response response) {
		ResponseFuture future = FUTURES.remove(response.getId());
		if (future != null) {
			future.doReceived(response);
		} else {
			logger.warn("The timeout response finally returned at "
					+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())) + ", response " + response
					+ response.getId());
		}
	}

	private void doReceived(Response res) {
		lock.lock();
		try {
			response = res;
			state.set(1);
			if (done != null) {
				done.signal();
			}
		} finally {
			lock.unlock();
		}
		if (callback != null) {
			invokeCallback(callback);
		}
	}

	@Override
	public Object get() throws InterruptedException, ExecutionException {
		lock.lock();
		try {
			while (!isDone()) {
				done.await();
				if (isDone()) {
					break;
				}
			}
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		} finally {
			lock.unlock();
		}
		return returnFromResponse();
	}

	@Override
	public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {

		if (timeout <= 0) {
			timeout = DEFAULT_TIMEOUT;
		}
		if (!isDone()) {
			long start = System.currentTimeMillis();
			lock.lock();
			try {
				while (!isDone()) {
					done.await(timeout, unit);
					if (isDone() || System.currentTimeMillis() - start > timeout) {
						break;
					}
				}
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			} finally {
				lock.unlock();
			}
			if (!isDone()) {
				throw new TimeoutException("读取超时：" + timeout + " ms");
			}
		}
		return returnFromResponse();
	}

	private Object returnFromResponse() {
		if (response == null) {
			throw new RemotingException("返回结果不能为null");
		}
		if (response.getThrowable() != null) {
			throw new RemotingException("调用失败", response.getThrowable());
		}
		return response.getResult();
	}

	private void invokeCallback(ResponseCallback c) {
		ResponseCallback callbackCopy = c;
		if (callbackCopy == null) {
			throw new RemotingException("回调对象不能为null");
		}
		c = null;
		Response res = response;
		if (res == null) {
			throw new RemotingException("结果不能为null");
		}

		if (state.get() == 2) {
			if (response.getThrowable() == null) {
				try {
					callbackCopy.done(res.getResult());
				} catch (Exception e) {
					logger.error("回调失败", e);
				}
			} else {
				callbackCopy.caught(new RuntimeException(response.getThrowable()));
			}

		} else {
			try {
				callbackCopy.caught(new TimeoutException());
			} catch (Exception e) {
				logger.error("回调失败", e);
			}
		}
	}

	public ResponseCallback getCallback() {
		return callback;
	}

	public void setCallback(ResponseCallback callback) {
		this.callback = callback;
		if (isDone()) {
			invokeCallback(callback);
		}
	}

	public Request getRequest() {
		return request;
	}

}
